package org.example.kafka.rebalance;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.example.kafka.common.CommonConstant;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 再平衡监听器的消费者
 */
public class RebalanceConsumer {
    public static final String GROUP_ID = "rebalanceconsumer";

    /**
     * 使用 ThreadFactoryBuilder 创建自定义线程名称的 ThreadFactory
     */
    private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
            .setNameFormat("hyn-demo-pool-%d").build();
    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(CommonConstant.CONCURRENT_PARTITIONS_COUNT,
            CommonConstant.CONCURRENT_PARTITIONS_COUNT,
            0L,
            TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<Runnable>(1000),
            namedThreadFactory,
            new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args)throws InterruptedException {

        for (int i = 0; i < CommonConstant.CONCURRENT_PARTITIONS_COUNT; i++) {
            executor.submit(new ConsumerWorker(false));
        }
        Thread.sleep(3000);
        //用来被停止，观察保持运行的消费者情况
        System.out.println("=====有新的消费者加入了======");
        new Thread(new ConsumerWorker(true)).start();
    }
}
